package omega

import (
	"context"
	"fmt"
	"reflect"

	"gitee.com/zawei/yggdrasil/pkg/lib/omega/proto"
	"gitee.com/zawei/yggdrasil/pkg/log"
	"github.com/pkg/errors"
)

type coordinate struct {
	funcs map[string]reflect.Value
}

func (p *coordinate) registerFunc(fnName string, fn interface{}) {
	p.funcs[fnName] = reflect.ValueOf(fn)
}

func (p *coordinate) executeCoordination(ctx context.Context, globalTxId, localTxId, parentTxId string, method string) error {
	cctx := context.WithValue(ctx, omegaCtxKey{}, &OmegaContext{
		GlobalTxId: globalTxId,
		LocalTxId:  localTxId,
		ParentTxId: parentTxId,
	})
	if v, ok := p.funcs[method]; ok {
		var results []reflect.Value
		if v.Type().IsVariadic() {
			results = v.CallSlice([]reflect.Value{reflect.ValueOf(cctx)})
		} else {
			results = v.Call([]reflect.Value{reflect.ValueOf(cctx)})
		}

		typeOfError := reflect.TypeOf((*error)(nil)).Elem()
		if len(results) > 0 {
			if results[len(results)-1].Type().Implements(typeOfError) && !results[len(results)-1].IsNil() {
				return results[len(results)-1].Interface().(error)
			}
		}
		abort, err := transportInstance.OnTccCoordinated(cctx, method, TransactionSucceed)
		if err != nil {
			return err
		}
		if abort {
			return errors.New(fmt.Sprintf("compensate %s : %s is aborted", globalTxId, localTxId))
		}
		return nil
	}

	return errors.Errorf("not tcc method: %s", method)
}

func (p *coordinate) schedule(ctx context.Context, cmdChannels []chan *proto.GrpcTccCoordinateCommand) {
	for i := 0; i < len(cmdChannels); i++ {
		go func(j int) {
			for {
				select {
				case cmd := <-cmdChannels[j]:
					log.Debugf("receive tcc command[%s]", log.Json(cmd))
					if err := p.executeCoordination(ctx, cmd.GlobalTxId, cmd.LocalTxId, cmd.ParentTxId, cmd.Method); err != nil {
						log.Errorf("fault to execute tcc command[%s], error: %s", log.Json(cmd), err.Error())
					}
				case <-ctx.Done():
					return
				}
			}
		}(i)
	}
}
